package cn.lecosa.spark.sort

import org.apache.spark.{ SparkContext, SparkConf }

object SecondarySortApp {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[*]")

    val sc = new SparkContext(conf)

    val lines = sc.textFile("D:\\JavaWorkspaces\\sparkproject\\sparktest.txt")

    val pairWithSortkey = lines.map(
        line => (  new SecondarySort(line.split(" ")(0).toInt, line.split(" ")(1).toInt), line)  )

    val sorted = pairWithSortkey.sortByKey(false)

    val sortedResult = sorted.map(sortedline => sortedline._2)
    sortedResult.collect.foreach(println)
  }
}

class SecondarySort(val first: Int, val second: Int) extends Ordered[SecondarySort] with Serializable {
  override def compare(that: SecondarySort): Int = {
    if (this.first - that.first != 0) {
      this.first - that.first
    } else {
      this.second - that.second
    }

  }
}